package com.mjk.common.core.dcc;


import com.mjk.common.tools.gson.GsonUtils;
import com.mjk.common.tools.logger.BizLogger;
import com.mjk.common.util.ObjectUtils;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.springframework.util.CollectionUtils;

import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * dcc 监控类 节点监控和节点数据监控
 * @author mjk
 *
 */
public class DccWatcher implements Watcher {
	/**
	 * 节点监控列表
	 */
	private final Map<String, NodeListener> nodeListeners = new ConcurrentHashMap<>();
	/**
	 * 节点数据监控
	 */
	private final Map<String, DataListener> dataListeners = new ConcurrentHashMap<>();
	
	private ZooKeeper zk;

	public DccWatcher(ZooKeeper zk) {
		this.zk = zk;
	}

	public void reconnect(ZooKeeper zk) {
		this.zk = zk;
		if(!CollectionUtils.isEmpty(dataListeners)) {
			for(Entry<String, NodeListener> item : nodeListeners.entrySet()){
				focusWatchNodeAndNotice(item.getKey(), item.getValue());
			}
			for(Entry<String,DataListener> item : dataListeners.entrySet()){
				focusWatchDataAndNotice(item.getKey(), item.getValue(), readData(item.getKey()));
			}
		}
	}

	@Override
	public void process(WatchedEvent event) {
		String path = event.getPath();//修改节点
		//事件类型
		switch (event.getType()) {
		case NodeCreated,NodeDeleted://节点创建
			for(Entry<String, NodeListener> item : nodeListeners.entrySet()){
				if (path.equals(item.getKey())) {
					focusWatchNodeAndNotice(path, nodeListeners.get(item.getKey()));
				}
			}
			break;
		case NodeChildrenChanged://节点子节点变化
			for(Entry<String, NodeListener> item : nodeListeners.entrySet()){
				if (path.equals(item.getKey())) {
					List<String> children = focusWatchNodeAndNotice(item.getKey(), item.getValue());
					item.getValue().onRefreshChildren(children);
				}
			}
			break;
		case NodeDataChanged://节点数据变化
			for(Entry<String, DataListener> entry:dataListeners.entrySet()) {	
				if (path.equals(entry.getKey())) {
					String data = readData(path);
					entry.getValue().onUpdate(path, data);
				}
			}
			break;
		default:
			break;
		}
	}
	
	public void addNodeListener(String path, NodeListener listener) {
		nodeListeners.put(path, listener);
		focusWatchNodeAndNotice(path, listener);
	}
	
	public void addDataListener(String path,DataListener listener) {
		dataListeners.put(path, listener);
	}
	
	public void removeNodeListener(String path){
		nodeListeners.remove(path);
	}
	
	public void removeDataListener(String path){
		dataListeners.remove(path);
	}
	
	public void delNode(String path) {
		try {
			this.zk.delete(path, -1);
		} catch (Exception e) {
			BizLogger.error(e.getMessage());
			Thread.currentThread().interrupt();
		}
	}

	public List<String> getChildren(String path) {
		try {
			return this.zk.getChildren(path, this);
		} catch (Exception e) {
			BizLogger.error(e.getMessage());
			Thread.currentThread().interrupt();
		}
		return Collections.emptyList();
	}
	
	private void focusWatchDataAndNotice(String path, DataListener listener, Object object) {
		if (object != null) {
			listener.onUpdate(path, object);
		}
	}
	/**
	 * 返回节点下面的子节点数据，并且给节点通知
     */
	private List<String> focusWatchNodeAndNotice(String watchPath, NodeListener listener) {
		try {
			if (!ObjectUtils.isEmpty(watchPath)) {
				Stat stat = this.zk.exists(watchPath, false);
				String nodeName = rightChindPath(watchPath);
				if (stat == null) {
					listener.onDestroy(nodeName);
				} else {
					listener.onConstruct(nodeName);
				}
			}
			if (this.zk.exists(watchPath, this) != null) {
				return this.zk.getChildren(watchPath, this);
			}
		} catch (Exception e) {
			BizLogger.error(watchPath + " " + e.getMessage());
			Thread.currentThread().interrupt();
		}
		return Collections.emptyList();
	}
	
	public String readData(String path) {
		try {
			byte[] data = zk.getData(path, this, null);
			return new String(data);
		} catch (Exception e) {
			BizLogger.error("读取数据异常{}", e);
			Thread.currentThread().interrupt();
		}
		return null;
	}
	
	public void writeData(String path, Object data,boolean isPersistent) {
		try {
			Stat stat = this.zk.exists(path, false);
			if (stat == null) {//创建了一个 永久节点
				if(isPersistent){
					this.zk.create(path, new byte[0],
						ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
				}else{//临时节点
					this.zk.create(path, new byte[0],
							ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);	
				}
			}
			zk.setData(path, Objects.requireNonNull(GsonUtils.bean2Json(data)).getBytes(), -1);
		} catch (Exception e) {
			BizLogger.error(e);
			Thread.currentThread().interrupt();
		}
	}	
	
	private String rightChindPath(String path) {
		if (path.startsWith("/")) {
			return path.substring(1);
		} else {
			return path;
		}
	}

	public interface NodeListener {

		/**
		 * 监听节点被删除时回调
		 */
		default void onDestroy(String nodeName){}

        /**
		 * <pre>
		 * 监听节点被创建时回调. 
		 * DccClient.registNodeListener("arc", listener);
		 * 当注册监听器时如果arc节点还不存在，监听器也会注册成功。在该节点被创建时会回调本方法
		 * </pre>
		 */
		default void onConstruct(String nodeName){}

        /**
         */
		default void onRefreshChildren(List<String> children){}
    }
	
	public interface DataListener {

		default void onUpdate(String path, Object data){}

    }
}
